package com.ds.lens.data.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2019-07-01
 */
@Component
@Slf4j
public class KafkaProducer {

    @Value("${lens.kafka.send.message.allowed}")
    private boolean allowedSend;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private KafkaTemplate<String, String> kafkaForecastTemplate;


    public void sendMessage(String topic, String key, String message) {
        if (allowedSend) {
            kafkaTemplate.send(topic, key, message);
        } else {
            log.error("The message not allowed to be sent! please check the properties:lens.kafka.send.message.allowed");
            log.error(String.format("topic:%s  | key:%s  | message:%s ", topic, key, message));
        }
    }

    public void sendMessageToForecast(String topic, String key, String message) {
        if (allowedSend) {
            kafkaForecastTemplate.send(topic, key, message);
        } else {
            log.error("The message not allowed to be sent! please check the properties:lens.kafka.send.message.allowed");
            log.error(String.format("topic:%s  | key:%s  | message:%s ", topic, key, message));
        }
    }

}
